package com.atguigu.userprofile.service.impl;

import com.alibaba.fastjson.JSON;
import com.atguigu.userprofile.bean.TagCondition;
import com.atguigu.userprofile.bean.TagInfo;
import com.atguigu.userprofile.bean.UserGroup;
import com.atguigu.userprofile.constants.ConstCodes;
import com.atguigu.userprofile.mapper.UserGroupMapper;
import com.atguigu.userprofile.service.TagInfoService;
import com.atguigu.userprofile.service.UserGroupService;
import com.atguigu.userprofile.utils.RedisUtil;
import com.baomidou.dynamic.datasource.annotation.DS;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;

import org.apache.catalina.User;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import redis.clients.jedis.Jedis;

import java.util.*;
import java.util.stream.Collectors;

/**
 * <p>
 * 服务实现类
 * </p>
 *
 * @author zhangchen
 * @since 2021-05-04
 */
@Service
@Slf4j
@DS("mysql")
public class UserGroupServiceImpl extends ServiceImpl<UserGroupMapper, UserGroup> implements UserGroupService
{


    /*
        1.生成必要的信息
            conditionJsonStr
            createTime
            conditionComment
        2.存入Mysql
     */
    @Override
    public void saveUserGroupMetaToMysql(UserGroup userGroup) {

        userGroup.setConditionJsonStr(JSON.toJSONString(userGroup.getTagConditions()));
        userGroup.setConditionComment(userGroup.conditionJsonToComment());
        userGroup.setCreateTime(new Date());

        save(userGroup);
    }

    /*
            计算分群，写入ck
     */
    @Override
    public void saveUserGroupToCk(UserGroup userGroup) {

        String sql = generateBitmapAndSql(userGroup);
        //调用Mapper
        super.baseMapper.insertUserGroupToCK(userGroup.getId().toString(),sql);

    }

    /*
            ck 中留底细
            redis 做缓存: 从ck中查询，写入redis

            --------------------------------
                在更新 分群信息时，从ck中查询，再写入redis。
                    保证从ck中查询数据的正确性(合并之后查询)。
                    如果ck尚未合并，需要手动合并！

                保证写入redis后的数据是正确的。
                        对于set集合来说，由于两次写入的key是一样的，因此会直接将不重复的部分写入set集合。
                        需要保证写入之前，先清空set集合！
     */
    @Override
    public Integer saveUserGroupToRedis(UserGroup userGroup) {

        String userGroupId = userGroup.getId().toString();
        //查询分群信息
        List<String> uids = baseMapper.queryUserGroupUsersFromCK(userGroupId);

        String [] value = {};
        value = uids.toArray(value);
        /*
            存的是某个分群的用户信息。

            写入redis
                key:    唯一标识。
                            usergroup_userGroupId
                value:
                        单个值: string,hash
                        多个值: list,set,hash,zset
                                选set更好。
                                    sismember快速判断一个元素是否在群体中。

         */
        Jedis jedis = RedisUtil.getJedis();

        String key = "usergroup_"+userGroupId;

        //先清空key 防止更新时数据不一致
        jedis.del(key);
        jedis.sadd(key,value);

        jedis.close();

        return uids.size();

    }

    /*
            根据条件，拼接一个 bitmapAnd的sql，查询返回结果即可
     */
    @Override
    public Integer evaluateUserGroupNum(UserGroup userGroup) {

        String sql = generateBitmapAndSql(userGroup);

        Integer num = baseMapper.evaluateUserGroupNum(sql);

        return num;
    }

    /*
         ck不支持常规意义的更新。
            先删除，再写入
     */
    @Override
    public void updateUserGroup(UserGroup userGroup) {

        String id = userGroup.getId().toString();
        //先删除
        baseMapper.deleteUserGroupById(id);
        //重新计算，写入
        String sql = generateBitmapAndSql(userGroup);
        baseMapper.insertUserGroupToCK(id,sql);

    }

    @Autowired
    private TagInfoService tagInfoService;
    /*
         生成 bitmapAnd 求交集的 sql表达式

    <select>： 不生成
       bitmapAnd(
         bitmapAnd(
            (sql1),(sql2)
         ),
         (sql3)
        )

        数量关系:  bitmapAnd = sql的数量 - 1
     */
    private String generateBitmapAndSql(UserGroup userGroup){

        //获取业务日期
        String doDate = userGroup.getBusiDate();
        //查询所有三级标签的元数据集合
        Map<String, TagInfo> tags = tagInfoService.getTagInfoMapWithCode();
        //获取所有的TagCondition
        List<TagCondition> tagConditions = userGroup.getTagConditions();

        StringBuilder result = new StringBuilder();

        // bitmapAnd( bitmapAnd( (sql1) ,(sql2 )) ,(sql3 ))
        for (TagCondition tagCondition : tagConditions) {

            String singleTagConditionSql = generateSingleTagConditionSql(tagCondition, doDate, tags);

            if (result.length() == 0){
                // 如果只有一个条件，需要在外层拼接 ()，否则和最外层的select拼接后，就会出错
                result.append("(").append(singleTagConditionSql).append(")");
            }else {
                result.insert(0,"bitmapAnd(").append(",(").append(singleTagConditionSql).append("))");
            }
        }

        return result.toString();
    }

    /*
        根据单个条件，生成查询的sql
       TagCondition(tagCode=TAG_POPULATION_ATTRIBUTE_NATURE_PERIOD, tagName=年代, operatorName=包含, operator=in,
       tagValues=[00后, 90后],
        tagCodePath=[TAG_POPULATION_ATTRIBUTE, TAG_POPULATION_ATTRIBUTE_NATURE, TAG_POPULATION_ATTRIBUTE_NATURE_PERIOD]
        ),

select groupBitmapMergeState(us)
                     from ①user_tag_value_string
                     where tag_code = ②'tag_population_attribute_nature_period'
                       and tag_value ③in ④('90后', '00后')
                       and dt = ⑤'2022-06-08'


                 ①根据查询条件中的标签，确定从 ck的哪张表查询
                        需要 根据 tagCode 获取 TagInfo,查询到标签值的类型。
                        调用 TagInfoService.getTagInfoMapWithCode 获取所有的三级标签的元数据信息。
                        再从Map中根据 根据 tagCode 获取 TagInfo

                 ②根据条件中的 tagCode 转为小写

                 ③操作符
                        需要把条件中的 operator转换为sql中的运算符
                                gt  ===>  >

                ④查询条件
                        根据 tagValues 生成。
                                如果是多个条件值，需要将条件中的每个条件值进行拼接
                                                [00后, 90后] ====>  ('90后', '00后')
                                                    条件值是 字符串或日期类型，需要添加 '
                                                    多个条件值直接，使用,间隔
                                                    多个条件，外层需要拼接 ()

                ⑤业务日期

     */
    public String generateSingleTagConditionSql(TagCondition tagCondition,String doDate,Map<String,TagInfo> tags){

        String template = " select groupBitmapMergeState(us)  from %s  where tag_code = '%s'" +
                          " and tag_value %s %s and dt = '%s' ";

        String tagCode = tagCondition.getTagCode();
        TagInfo tagInfo = tags.get(tagCode);

        //确定查询的表名
        String sourceTable = "";
        switch (tagInfo.getTagValueType()){
            case ConstCodes.TAG_VALUE_TYPE_LONG:  sourceTable = "user_tag_value_long"; break;
            case ConstCodes.TAG_VALUE_TYPE_DECIMAL:  sourceTable = "user_tag_value_decimal"; break;
            case ConstCodes.TAG_VALUE_TYPE_STRING:  sourceTable = "user_tag_value_string";break;
            case ConstCodes.TAG_VALUE_TYPE_DATE:  sourceTable = "user_tag_value_date"; break;
        }

        //需要把条件中的 operator转换为sql中的运算符
        String sqlOperator = getConditionOperator(tagCondition.getOperator());

        // 根据 tagValues 生成,如果是多个条件值，需要将条件中的每个条件值进行拼接
        String tagValueSql = "";
        if (ConstCodes.TAG_VALUE_TYPE_DATE.equals(tagInfo.getTagValueType())
            ||
            ConstCodes.TAG_VALUE_TYPE_STRING.equals(tagInfo.getTagValueType())
        ){
            tagValueSql = tagCondition.getTagValues().stream().map(v -> "'" + v + "'").collect(Collectors.joining(","));
        }else{
            tagValueSql = StringUtils.join(tagCondition.getTagValues(),",");
        }
        //如果是 in 或 notin ，需要在tagValues的外层拼接 ()
        if ("in".equals(tagCondition.getOperator()) || "nin".equals(tagCondition.getOperator())){
            tagValueSql = "(" + tagValueSql + ")";
        }

        //格式化
        String sql = String.format(template, sourceTable, tagCode.toLowerCase(), sqlOperator, tagValueSql, doDate);

        System.out.println(sql);

        return sql;
    }

    //读需求说明
    private String getConditionOperator(String operator){

        switch (operator) {
            case "eq" : return "=";
            case "neq" : return "<>";
            case "gte" : return ">=";
            case "lte" : return "<=";
            case "gt" : return ">";
            case "lt" : return "<";
            case "in" : return "in";
            case "nin" : return "not in";
        }
        throw new RuntimeException("没有匹配的操作符!");

    }
}
